package com.etc

import com.etc.common.{Constants, TopicName}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.{PreferBrokers, PreferConsistent, PreferFixed}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * @Author kalista
 * @Description
 * @Date 2021/5/23  14:31
 * */
object OrderStreamingProcessor {


  def main(args: Array[String]): Unit = {

    // 设置spark程序在控制台中得日志打印级别
    Logger.getLogger("org").setLevel(Level.WARN)
    //local[*] 使用本地模式运行
    val conf = new SparkConf().setAppName("orderMonitor").setMaster("local[*]")
    // 初始化sparkstreaming
    val streamingContext = new StreamingContext(conf,Seconds(1))
    // 设置检查点  checkpoint
    streamingContext.checkpoint("/sparkapp/tmp")

    // 构建kafka连接得map
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "hdp-01:9092,hdp-02:9092,hdp-03:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "test0002",
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )

    val topics = Array(
      TopicName.HAI_KOU_ORDER_TOPIC.getName,
      TopicName.cheng_du_ORDER_TOPIC.getName,
      TopicName.xi_an_ORDER_TOPIC.getName)
//     val topics = Array("hai_kou_gps_topic")

    val stream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      streamingContext,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )

    // 实时统计订单总数
    val orderDS = stream.map(record =>{
      // 主题名称
      val topicName = record.topic()
      val orderInfo = record.value()

      var orderParser:OrderParser = null

      topicName match {
        case "hai_kou_order_topic" =>{
          orderParser = new HaiKouOrderParser();
        }
        case "cheng_du_order_topic" =>{
          orderParser = new ChengDuOrderParser()
        }
        case "xi_an_order_topic" =>{
          orderParser = new XiAnOrderParser()
        }
        case _ =>{
          orderParser = null
        }
      }
      println("orderparser:" + orderParser)

      if (null != orderParser){
        val order = orderParser.parser(orderInfo)
        println("order: " + order)
        order
      }else{
        null
      }
    })

    // 订单计数   对于每个订单出现一次计数为1
    // (城市编码_订单天,1)

    val orderCountRest = orderDS.map(order => {
      if (null == order) {
        ("", 0)
      } else if (order.getClass == classOf[ChengDuTravelOrder]) {
        (Constants.CITY_CODE_CHENG_DU + "_" + order.createDay, 1)
      } else if (order.getClass == classOf[XiAnTravelOrder]) {
        (Constants.CITY_CODE_XI_AN + "_" + order.createDay, 1)
      } else if (order.getClass == classOf[HaiKouTravelOrder]) {
        (Constants.CITY_CODE_HAI_KOU + "_" + order.createDay, 1)
      } else {
        ("", 0)
      }
    }).updateStateByKey((currValues: Seq[Int], state: Option[Int]) => {
      var count = currValues.sum + state.getOrElse(0)
      Some(count)
    })



    // 乘车人数统计  passengerCount 字段  如果有就累加   如果没有就认为一个人
    val passengerCountRest = orderDS.map(order => {
      if (null == order) {
        ("", 0)
      } else if (order.getClass == classOf[ChengDuTravelOrder]) {
        (Constants.CITY_CODE_CHENG_DU + "_" + order.createDay, 1)
      } else if (order.getClass == classOf[XiAnTravelOrder]) {
        (Constants.CITY_CODE_XI_AN + "_" + order.createDay, 1)
      } else if (order.getClass == classOf[HaiKouTravelOrder]) {
        // 获取人数
        var passengerCount = order.asInstanceOf[HaiKouTravelOrder].passengerCount.toInt
        // scala 不支持 类似java 三目运算符  可以使用下面
        passengerCount = if (passengerCount >0) passengerCount else 1
        (Constants.CITY_CODE_XI_AN + "_" + order.createDay, passengerCount)
      } else {
        ("", 0)
      }
    }).updateStateByKey((currValues: Seq[Int], state: Option[Int]) => {
      var count = currValues.sum + state.getOrElse(0)
      Some(count)
    })


    // 保存订单数据到redis中
    orderCountRest.foreachRDD(orderCountRDD =>{
      import  com.etc.util.JedisUtil
      // 获取到redis 连接
      val jedisUtil = JedisUtil.getInstance()
      val jedis = jedisUtil.getJedis
      // 从集群中收集到所有得数据放到redis
      val orderCount: Array[(String, Int)] = orderCountRDD.collect()
      print("orderCountRest:"+orderCount)
      orderCount.foreach( countrest =>{
        println("countrest:"+countrest._1+","+countrest._2)
        if (null != countrest){
          jedis.hset(Constants.ORDER_COUNT,countrest._1,countrest._2+"")
        }
      })
      jedisUtil.returnJedis(jedis)
    })


    // 乘车人数保存redis中
    passengerCountRest.foreachRDD(passengerCount =>{
      import  com.etc.util.JedisUtil
      // 获取到redis 连接
      val jedisUtil = JedisUtil.getInstance()
      val jedis = jedisUtil.getJedis
      // 从集群中收集到所有得数据放到redis
      val passengerCountRestt: Array[(String, Int)] = passengerCount.collect()
      print("passengerCountRestt:"+passengerCountRestt)
      passengerCountRestt.foreach( countrest =>{
        println("passengerCountRestt:"+countrest._1+","+countrest._2)
        if (null != countrest){
          jedis.hset(Constants.PASSENGER_COUNT,countrest._1,countrest._2+"")
        }
      })
      jedisUtil.returnJedis(jedis)
    })


    /// 启动sparkstreming程序
    streamingContext.start()
    streamingContext.awaitTermination()
    streamingContext.stop()



  }



}
